from absl import app, flags
import pandas as pd
import toml
import time

from coin.base.query_util import query_exchange_tickers
from coin.util.queue.config import KafkaConfig
from coin.support.market_quotes.logic.constants import TICKER_CHECK_LIST
from coin.support.market_quotes.logic.kafka_querier import KafkaMarketInfoQuerier
from xunkemgmt_client.support.accounting.logic.constants import FIAT_LIST
from xunkemgmt_client.tool.slack_noti import send_to_slack
from coin.proto.coin_request_pb2 import AccountRequestProto
from google.protobuf.json_format import MessageToDict

FLAGS = flags.FLAGS


def check_exchange_tickers():
  symbol_group_file = f'data/coin2/feed/symbol_groups.toml'
  with open(symbol_group_file) as f:
    symbol_groups_universe = toml.load(f)
  exchange_ticker_list = []
  for mea in TICKER_CHECK_LIST:
    market_type, exchange, api_version = mea.split('.')
    ex_tickers = query_exchange_tickers(market_type=market_type,
                                        exchange=exchange,
                                        api_version=api_version)
    num_tickers = len(ex_tickers.each_ticker) if ex_tickers else 0
    if exchange == 'ExchangeRates':
      num_tickers_needed = len(FIAT_LIST)
    else:
      symbol_groups = [*(symbol_groups_universe[mea.lower().replace('.', '_')].values())]
      num_tickers_needed = len([symbol for symbols in symbol_groups for symbol in symbols])
    exchange_ticker_list.append((market_type, exchange, api_version, 
                                 num_tickers / num_tickers_needed))
  return exchange_ticker_list


def _convert_ticker_to_df(ticker_list):
  if not ticker_list:
    return pd.DataFrame()
  ticker_list = [ticker.ticker for ticker in ticker_list if ticker is not None]
  new_ticker_list = []
  for ticker_info in ticker_list:
    mea = {
      'market_type': ticker_info.market_type,
      'exchange': ticker_info.exchange,
      'api_version': ticker_info.api_version}
    for each_ticker in ticker_info.each_ticker:
      each_ticker = MessageToDict(each_ticker, preserving_proto_field_name=True)
      new_ticker_list.append({**mea, **each_ticker})
  ticker_df = pd.DataFrame(new_ticker_list)
  return ticker_df


def check_coin_price(kafka_config_filename, offset_sec):
  kafka_config = KafkaConfig.from_cmd_config(kafka_config_filename)
  querier = KafkaMarketInfoQuerier(kafka_config)
  request_list = []
  for mea in TICKER_CHECK_LIST:
    market_type, exchange, api_version = mea.split('.')
    request_list.append(AccountRequestProto(
      market_type=market_type, exchange=exchange, api_version=api_version))
  ticker_df = _convert_ticker_to_df(querier.query_ticker_history(request_list))
  if FLAGS.exclude_symbol_regex:
    ticker_df = ticker_df[~ticker_df['symbol'].str.contains(FLAGS.exclude_symbol_regex)]
  if len(ticker_df) == 0:
    return ticker_df
  query_timestamp = int((time.time() - offset_sec) * 10 ** 9)
  prev_ticker_df = _convert_ticker_to_df(
    querier.query_ticker_history(request_list, query_timestamp))
  compare_df = prev_ticker_df.merge(ticker_df,
    on=['market_type', 'exchange', 'api_version', 'symbol'], suffixes=('_before', ''))
  compare_df['ratio'] = compare_df['last_before'] / compare_df['last']
  compare_df = compare_df[(compare_df['ratio'] >= 10) | (compare_df['ratio'] <= 0.1)]
  return compare_df


def main(_):
  exchange_ticker_list = check_exchange_tickers()
  exchange_ticker_df = pd.DataFrame(exchange_ticker_list,
                                    columns=['market_type', 'exchange','api_version', 'ticker_ratio'])
  coin_price_df = check_coin_price(FLAGS.kafka_config_filename, FLAGS.price_offset_sec)
  print(exchange_ticker_df.to_string())
  if FLAGS.complete_thold:
    exchange_ticker_df = exchange_ticker_df[
      exchange_ticker_df['ticker_ratio'] < FLAGS.complete_thold]
    exchange_ticker_df['ticker_ratio'] = exchange_ticker_df['ticker_ratio'].apply(
      lambda x: "{:.2f}".format(x))
  if FLAGS.slack_receiver:
    mention_list = ['chensili', 'dongzhang']
    noti_msg = ""
    if len(exchange_ticker_df) > 0:
      noti_msg += "ticker exchange warning:\n"
      noti_msg += exchange_ticker_df.to_string(index=False)
    if len(coin_price_df) > 0:
      noti_msg += "\nabnormal coin price warning:\n"
      noti_msg += coin_price_df.to_string(index=False)
    if noti_msg:
      send_to_slack("```" + noti_msg + "```",
      FLAGS.slack_receiver, 'msg', mention_list=mention_list)


if __name__ == '__main__':
  flags.DEFINE_string('kafka_config_filename',
                      '../../coin_deploy/support_monitor/config/kafka_aws_config_2.json',
                      'kafka config file path')
  flags.DEFINE_float('complete_thold', 0.8, 'exchange tickers completeness thold')
  flags.DEFINE_integer('price_offset_sec', 1800, 'price check offset second')
  flags.DEFINE_string('slack_receiver', 'hft_raw_noti', '')
  flags.DEFINE_string('exclude_symbol_regex', None, 'exclude check ticker regex')
  app.run(main)
